package com.nd.bdc.presto.service;

import com.nd.bdc.presto.entity.ExecuteMessage;
import com.nd.bdc.presto.entity.PrestoSelectResult;
import com.nd.bdc.presto.websocket.MessageDisruptorQueue;
import io.prestosql.jdbc.QueryStats;
import org.springframework.stereotype.Component;
import java.io.Serializable;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * <p>Description:  presto的执行sql过程中的数据返回（通过websocket返回）</p>
 *
 * @author ggh
 * @date 2020/6/4
 */
@Component
public class PrestoWebSocketService {


    /**
     * 发送执行开始消息。
     *
     * @param channel
     * @param sqlCount
     */
    public void sendStartMessage(String channel, int sqlCount)
    {
        MessageData messageData = new StartMessageData(sqlCount);
        MessageDisruptorQueue.publishEvent(channel,messageData);
    }

    /**
     * 发送SQL执行成功消息。
     *
     * @param channel
     * @param sqlStatement
     */
    public void sendSqlSuccessMessage(String channel, String sqlStatement, int sqlStatementIndex)
    {
        MessageData messageData = new SqlSuccessMessageData(sqlStatement, sqlStatementIndex, SqlResultType.NONE);
        MessageDisruptorQueue.publishEvent(channel,messageData);
    }

    /**
     * 发送SQL执行成功消息。
     *
     * @param channel
     * @param sqlStatement
     * @param sqlStatementIndex
     * @param updateCount
     */
    public void sendSqlSuccessMessage(String channel, String sqlStatement, int sqlStatementIndex,
                                      int updateCount)
    {
        SqlSuccessMessageData sqlSuccessMessageData = new SqlSuccessMessageData(sqlStatement, sqlStatementIndex,
                SqlResultType.UPDATE_COUNT);
        sqlSuccessMessageData.setUpdateCount(updateCount);
        MessageDisruptorQueue.publishEvent(channel,sqlSuccessMessageData);
    }

    /**
     * 发送SQL执行成功消息。
     *
     * @param channel
     * @param sqlStatement
     * @param sqlStatementIndex
     * @param sqlSelectResult
     */
    public void sendSqlSuccessMessage(String channel, String sqlStatement, int sqlStatementIndex,
                                      PrestoSelectResult sqlSelectResult)
    {
        SqlSuccessMessageData sqlSuccessMessageData = new SqlSuccessMessageData(sqlStatement, sqlStatementIndex,
                SqlResultType.RESULT_SET);
        sqlSuccessMessageData.setSqlSelectResult(sqlSelectResult);
        MessageDisruptorQueue.publishEvent(channel,sqlSuccessMessageData);
    }

    /**
     * 发送SQL执行监控消息。
     *
     * @param channel
     * @param sqlStatement
     * @param sqlStatementIndex
     * @param queryStats
     */

    public void sendProgressMonitorMessage(String channel, String sqlStatement, int sqlStatementIndex,
                                      QueryStats queryStats)
    {
        ProgressMonitorMessageData progressMonitorMessageData = new ProgressMonitorMessageData(sqlStatement, sqlStatementIndex,
                queryStats);
        MessageDisruptorQueue.publishEvent(channel,progressMonitorMessageData);
    }

    /**
     * 发送SQL异常消息。
     *
     * @param channel
     * @param sqlStatement
     * @param sqlStatementIndex
     * @param e
     * @param content
     */
    public void sendSqlExceptionMessage(String channel, String sqlStatement, int sqlStatementIndex,
                                        SQLException e, String content)
    {
        SQLExceptionMessageData messageData = new SQLExceptionMessageData(sqlStatement, sqlStatementIndex, content + ":" + e.getCause().getMessage());
        MessageDisruptorQueue.publishEvent(channel, messageData);
    }

    /**
     * 发送SQL异常消息。
     *
     * @param channel
     * @param sqlStatement
     * @param sqlStatementIndex
     * @param content
     */
    public void sendSqlExceptionMessage(String channel, String sqlStatement, int sqlStatementIndex,
                                        String content)
    {
        SQLExceptionMessageData messageData = new SQLExceptionMessageData(sqlStatement, sqlStatementIndex, content);
        MessageDisruptorQueue.publishEvent(channel, messageData);
    }

    /**
     * 发送异常消息。
     *
     * @param channel
     * @param t
     * @param content
     * @param trace
     */
    public void sendExceptionMessage(String channel, Throwable t, String content, boolean trace)
    {
        ExceptionMessageData messageData = new ExceptionMessageData(content);
        if (trace) {
            messageData.setDetailTrace(t);
        }
        MessageDisruptorQueue.publishEvent(channel, messageData);
    }

    /**
     * 发送异常消息。
     *
     * @param channel
     * @param content
     */
    public void sendExceptionMessage(String channel, String content)
    {
        ExceptionMessageData messageData = new ExceptionMessageData(content);
        MessageDisruptorQueue.publishEvent(channel, messageData);
    }


    /**
     * 发送文本消息。
     *
     * @param channel
     * @param text
     */
    public void sendTextMessage(String channel, String text)
    {
        MessageData messageData = new TextMessageData(text);
        MessageDisruptorQueue.publishEvent(channel,messageData);
    }

    /**
     * 发送文本消息。
     *
     * @param channel
     * @param text
     * @param cssClass
     */
    public void sendTextMessage(String channel, String text, String cssClass)
    {
        TextMessageData textMessageData = new TextMessageData(text);
        textMessageData.setCssClass(cssClass);
        MessageDisruptorQueue.publishEvent(channel, textMessageData);
    }

    /**
     * 发送文本消息。
     *
     * @param channel
     * @param text
     * @param cssClass
     * @param sqlExecutionStat
     */
    public void sendTextMessage(String channel, String text, String cssClass, SQLExecutionStat sqlExecutionStat)
    {
        TextMessageData textMessageData = new TextMessageData(text);
        textMessageData.setCssClass(cssClass);
        textMessageData.setSqlExecutionStat(sqlExecutionStat);

        MessageDisruptorQueue.publishEvent(channel, textMessageData);
    }

    /**
     * 发送执行完成消息。
     * <p>
     * 无论是否出现异常，都要发送此消息。
     * </p>
     *
     * @param channel
     */
    public void sendFinishMessage(String channel)
    {
        FinishMessageData finishMessageData =new FinishMessageData();
        MessageDisruptorQueue.publishEvent(channel, finishMessageData);
    }

    /**
     * 发送执行完成消息。
     * <p>
     * 无论是否出现异常，都要发送此消息。
     * </p>
     *
     * @param channel
     * @param sqlExecutionStat
     */
    public void sendFinishMessage(String channel, SQLExecutionStat sqlExecutionStat)
    {
        FinishMessageData finishMessageData = new FinishMessageData();
        finishMessageData.setSqlExecutionStat(sqlExecutionStat);
        MessageDisruptorQueue.publishEvent(channel, finishMessageData);
    }

    /**
     * 获取指定ID的消息通道。
     *
     * @param channelId
     * @return
     */
    public String getChannel(String channelId)
    {
        return null;
    }

    /**
     * 获取指定ID的消息通道，必要时建立通道。
     *
     * @param channelId
     * @return
     */
    public String getChannelWithCreation(String channelId)
    {
        return null;
    }


    public static abstract class MessageData
    {
        protected static final String TIME_PATTERN = "HH:mm:ss";

        private String type;

        private Date date;

        public MessageData()
        {
            super();
            this.date = new Date();
        }

        public MessageData(String type)
        {
            super();
            this.type = type;
            this.date = new Date();
        }

        public String getType()
        {
            return type;
        }

        protected void setType(String type)
        {
            this.type = type;
        }

        public Date getDate()
        {
            return date;
        }

        public void setDate(Date date)
        {
            this.date = date;
        }

        public String getTimeText()
        {
            return new SimpleDateFormat(TIME_PATTERN).format(this.date);
        }
    }

    protected static class StartMessageData extends MessageData
    {
        public static final String TYPE = "START";

        private int sqlCount = 0;

        public StartMessageData()
        {
            super(TYPE);
        }

        public StartMessageData(int sqlCount)
        {
            super(TYPE);
            this.sqlCount = sqlCount;
        }

        public int getSqlCount()
        {
            return sqlCount;
        }

        public void setSqlCount(int sqlCount)
        {
            this.sqlCount = sqlCount;
        }
    }

    protected static class SqlSuccessMessageData extends MessageData
    {
        public static final String TYPE = "SQLSUCCESS";

        private String sqlStatement;

        /** SQL语句索引 */
        private int sqlStatementIndex;

        /** SQL结果类型 */
        private SqlResultType sqlResultType = SqlResultType.NONE;

        /** 更新数目 */
        private int updateCount = -1;

        private PrestoSelectResult sqlSelectResult;

        public SqlSuccessMessageData()
        {
            super(TYPE);
        }

        public SqlSuccessMessageData(String sqlStatement, int sqlStatementIndex, SqlResultType sqlResultType)
        {
            super(TYPE);
            this.sqlStatement = sqlStatement;
            this.sqlStatementIndex = sqlStatementIndex;
            this.sqlResultType = sqlResultType;
        }

        public String getSqlStatement()
        {
            return sqlStatement;
        }

        public void setSqlStatement(String sqlStatement)
        {
            this.sqlStatement = sqlStatement;
        }

        public int getSqlStatementIndex()
        {
            return sqlStatementIndex;
        }

        public void setSqlStatementIndex(int sqlStatementIndex)
        {
            this.sqlStatementIndex = sqlStatementIndex;
        }

        public SqlResultType getSqlResultType()
        {
            return sqlResultType;
        }

        public void setSqlResultType(SqlResultType sqlResultType)
        {
            this.sqlResultType = sqlResultType;
        }

        public int getUpdateCount()
        {
            return updateCount;
        }

        public void setUpdateCount(int updateCount)
        {
            this.updateCount = updateCount;
        }

        public PrestoSelectResult getSqlSelectResult()
        {
            return sqlSelectResult;
        }

        public void setSqlSelectResult(PrestoSelectResult sqlSelectResult)
        {
            this.sqlSelectResult = sqlSelectResult;
        }
    }

    protected static class ExceptionMessageData extends MessageData
    {
        public static final String TYPE = "EXCEPTION";

        private String content;

        private String detailTrace;

        public ExceptionMessageData()
        {
            super(TYPE);
        }

        public ExceptionMessageData(String content)
        {
            super(TYPE);
            this.content = content;
        }

        public String getContent()
        {
            return content;
        }

        public void setContent(String content)
        {
            this.content = content;
        }

        public String getDetailTrace()
        {
            return detailTrace;
        }

        public void setDetailTrace(String detailTrace)
        {
            this.detailTrace = detailTrace;
        }

        public void setDetailTrace(Throwable t)
        {
            this.detailTrace = ExecuteMessage.printThrowableTrace(t);
        }
    }

    protected static class SQLExceptionMessageData extends ExceptionMessageData
    {
        public static final String TYPE = "SQLEXCEPTION";

        private String sqlStatement;

        /** SQL语句索引 */
        private int sqlStatementIndex;

        public SQLExceptionMessageData()
        {
            super();
            super.setType(TYPE);
        }

        public SQLExceptionMessageData(String sqlStatement, int sqlStatementIndex, String content)
        {
            super(content);
            super.setType(TYPE);
            this.sqlStatement = sqlStatement;
            this.sqlStatementIndex = sqlStatementIndex;
        }

        public String getSqlStatement()
        {
            return sqlStatement;
        }

        public void setSqlStatement(String sqlStatement)
        {
            this.sqlStatement = sqlStatement;
        }

        public int getSqlStatementIndex()
        {
            return sqlStatementIndex;
        }

        public void setSqlStatementIndex(int sqlStatementIndex)
        {
            this.sqlStatementIndex = sqlStatementIndex;
        }
    }

    protected static class ProgressMonitorMessageData extends MessageData
    {
        public static final String TYPE = "MONITOR";

        private String sqlStatement;

        /** SQL语句索引 */
        private int sqlStatementIndex;

        private QueryStats queryStats;

        public ProgressMonitorMessageData()
        {
            super();
            super.setType(TYPE);
        }

        public ProgressMonitorMessageData(String sqlStatement, int sqlStatementIndex, QueryStats queryStats)
        {
            super.setType(TYPE);
            this.sqlStatement = sqlStatement;
            this.sqlStatementIndex = sqlStatementIndex;
            this.queryStats = queryStats;
        }

        public String getSqlStatement()
        {
            return sqlStatement;
        }

        public void setSqlStatement(String sqlStatement)
        {
            this.sqlStatement = sqlStatement;
        }

        public int getSqlStatementIndex()
        {
            return sqlStatementIndex;
        }

        public void setSqlStatementIndex(int sqlStatementIndex)
        {
            this.sqlStatementIndex = sqlStatementIndex;
        }

        public QueryStats getQueryStats() {
            return queryStats;
        }

        public void setQueryStats(QueryStats queryStats) {
            this.queryStats = queryStats;
        }
    }


    protected static class TextMessageData extends MessageData
    {
        public static final String TYPE = "TEXT";

        private String text;

        private String cssClass;

        private SQLExecutionStat sqlExecutionStat;

        public TextMessageData()
        {
            super(TYPE);
        }

        public TextMessageData(String text)
        {
            super(TYPE);
            this.text = text;
        }

        public String getText()
        {
            return text;
        }

        public void setText(String text)
        {
            this.text = text;
        }

        public String getCssClass()
        {
            return cssClass;
        }

        public void setCssClass(String cssClass)
        {
            this.cssClass = cssClass;
        }

        public SQLExecutionStat getSqlExecutionStat()
        {
            return sqlExecutionStat;
        }

        public void setSqlExecutionStat(SQLExecutionStat sqlExecutionStat)
        {
            this.sqlExecutionStat = sqlExecutionStat;
        }
    }

    protected static class FinishMessageData extends MessageData
    {
        public static final String TYPE = "FINISH";

        private SQLExecutionStat sqlExecutionStat;

        public FinishMessageData()
        {
            super(TYPE);
        }

        public SQLExecutionStat getSqlExecutionStat()
        {
            return sqlExecutionStat;
        }

        public void setSqlExecutionStat(SQLExecutionStat sqlExecutionStat)
        {
            this.sqlExecutionStat = sqlExecutionStat;
        }
    }

    /**
     * SQL执行结果类型。
     *
     * @author ggh
     *
     */
    public static enum SqlResultType
    {
        /** 结果集 */
        RESULT_SET,

        /** 更新数目 */
        UPDATE_COUNT,

        /** 无结果 */
        NONE
    }

    public static class SQLExecutionStat implements Serializable
    {
        private static final long serialVersionUID = 1L;

        /** 总计SQL语句数 */
        private int totalCount;

        /** 执行成功数 */
        private int successCount = 0;

        /** 执行失败数 */
        private int exceptionCount = 0;

        /** SQL执行持续毫秒数，-1表示未记录 */
        private long sqlDuration = -1;

        /** 任务执行持续毫秒数，-1表示未记录 */
        private long taskDuration = -1;

        public SQLExecutionStat()
        {
            super();
        }

        public SQLExecutionStat(int totalCount)
        {
            super();
            this.totalCount = totalCount;
        }

        public SQLExecutionStat(int totalCount, int successCount, int exceptionCount, long sqlDuration)
        {
            super();
            this.totalCount = totalCount;
            this.successCount = successCount;
            this.exceptionCount = exceptionCount;
            this.sqlDuration = sqlDuration;
        }

        public int getTotalCount()
        {
            return totalCount;
        }

        public void setTotalCount(int totalCount)
        {
            this.totalCount = totalCount;
        }

        public int getSuccessCount()
        {
            return successCount;
        }

        public void setSuccessCount(int successCount)
        {
            this.successCount = successCount;
        }

        public int getExceptionCount()
        {
            return exceptionCount;
        }

        public void setExceptionCount(int exceptionCount)
        {
            this.exceptionCount = exceptionCount;
        }

        public long getSqlDuration()
        {
            return sqlDuration;
        }

        public void setSqlDuration(long sqlDuration)
        {
            this.sqlDuration = sqlDuration;
        }

        public long getTaskDuration()
        {
            return taskDuration;
        }

        public void setTaskDuration(long taskDuration)
        {
            this.taskDuration = taskDuration;
        }

        public int getAbortCount()
        {
            return this.totalCount - this.successCount - this.exceptionCount;
        }

        public void increaseSuccessCount()
        {
            this.successCount += 1;
        }

        public void increaseExceptionCount()
        {
            this.exceptionCount += 1;
        }

        public void increaseSqlDuration(long increment)
        {
            if (this.sqlDuration < 0) {
                this.sqlDuration = 0;
            }

            this.sqlDuration += increment;
        }

        public void increaseTaskDuration(long increment)
        {
            if (this.taskDuration < 0) {
                this.taskDuration = 0;
            }

            this.taskDuration += increment;
        }
    }



}
